
#include <gtest/gtest.h>

#include <xbase_utility.hpp>
#include "../tablet_server/sstable.hpp"

using namespace xbase;

// 检查刚创建后，各个接口是否符合预期
TEST(test_sstable, init_creating_sstable)
{
	sstable_block_cache_manager::init_instance(5120);

	{
		sstable::pointer p = sstable::create("/tmp/init_creating_sstable",
						     "/tmp/init_creating_sstable.compacting",
						     "/tmp/init_creating_sstable.deleted",
						     "00000",
						     "11111",
						     0,
						     64 * 1024,
						     compressor::NONE);
		ASSERT_TRUE(p);

		ASSERT_EQ(0, p->sequence_number());
		ASSERT_EQ("", p->get_split_row());
		ASSERT_EQ("/tmp/init_creating_sstable", p->path());

		ASSERT_EQ(0, p->file_size());
		ASSERT_EQ(false, p->is_ref_sstable());
		ASSERT_EQ(false, p->is_ref_top_sstable());
		ASSERT_EQ(false, p->is_ref_bottom_sstable());

		{
			sstable_block_cache block;
			block_entry entry;
			std::pair<std::size_t, std::size_t> pair(0,0);
			ASSERT_FALSE(p->get_entry(block, entry, pair));
		}

		ASSERT_FALSE(fs::exists("/tmp/init_creating_sstable"));
		ASSERT_TRUE(fs::exists("/tmp/init_creating_sstable.compacting"));
	}
	ASSERT_FALSE(fs::exists("/tmp/init_creating_sstable"));
	ASSERT_FALSE(fs::exists("/tmp/init_creating_sstable.compacting"));
	ASSERT_FALSE(fs::exists("/tmp/init_creating_sstable.deleted"));

	sstable_block_cache_manager::reset_instance();
}

// 检查刚加载后，各个接口是否符合预期
TEST(test_sstable, init_loading_sstable)
{
}

TEST(test_sstable, half_created_sstable)
{
	sstable_block_cache_manager::init_instance(5120);

	{
		sstable::pointer p = sstable::create("/tmp/half_created_sstable",
						     "/tmp/half_created_sstable.compacting",
						     "/tmp/half_created_sstable.deleted",
						     "00000",
						     "11111",
						     0,
						     64 * 1024,
						     compressor::NONE);
		ASSERT_TRUE(p);

		ASSERT_EQ(0, p->sequence_number());
		ASSERT_EQ("", p->get_split_row());
		ASSERT_EQ("/tmp/half_created_sstable", p->path());

		ASSERT_EQ(0, p->file_size());
		ASSERT_EQ(false, p->is_ref_sstable());
		ASSERT_EQ(false, p->is_ref_top_sstable());
		ASSERT_EQ(false, p->is_ref_bottom_sstable());

		{
			sstable_block_cache block;
			block_entry entry;
			std::pair<std::size_t, std::size_t> pair(0,0);
			ASSERT_FALSE(p->get_entry(block, entry, pair));
		}

		ASSERT_FALSE(fs::exists("/tmp/half_created_sstable"));
		ASSERT_TRUE(fs::exists("/tmp/half_created_sstable.compacting"));

		// 模拟写失败
	}
	ASSERT_FALSE(fs::exists("/tmp/half_created_sstable"));
	ASSERT_FALSE(fs::exists("/tmp/half_created_sstable.compacting"));
	ASSERT_FALSE(fs::exists("/tmp/half_created_sstable.deleted"));

	sstable_block_cache_manager::reset_instance();
}

TEST(test_sstable, half_created_sstable_with_deleted)
{
	sstable_block_cache_manager::init_instance(5120);

	{
		sstable::pointer p = sstable::create("/tmp/half_created_sstable_with_deleted",
						     "/tmp/half_created_sstable_with_deleted.compacting",
						     "/tmp/half_created_sstable_with_deleted.deleted",
						     "00000",
						     "11111",
						     0,
						     64 * 1024,
						     compressor::NONE);
		ASSERT_TRUE(p);

		ASSERT_EQ(0, p->sequence_number());
		ASSERT_EQ("", p->get_split_row());
		ASSERT_EQ("/tmp/half_created_sstable_with_deleted", p->path());

		ASSERT_EQ(0, p->file_size());
		ASSERT_EQ(false, p->is_ref_sstable());
		ASSERT_EQ(false, p->is_ref_top_sstable());
		ASSERT_EQ(false, p->is_ref_bottom_sstable());

		{
			sstable_block_cache block;
			block_entry entry;
			std::pair<std::size_t, std::size_t> pair(0,0);
			ASSERT_FALSE(p->get_entry(block, entry, pair));
		}

		ASSERT_FALSE(fs::exists("/tmp/half_created_sstable_with_deleted"));
		ASSERT_TRUE(fs::exists("/tmp/half_created_sstable_with_deleted.compacting"));

		// 模拟写失败
		ASSERT_TRUE(p->set_deleted());
		ASSERT_TRUE(fs::exists("/tmp/half_created_sstable_with_deleted.deleted"));
	}
	ASSERT_FALSE(fs::exists("/tmp/half_created_sstable_with_deleted"));
	ASSERT_FALSE(fs::exists("/tmp/half_created_sstable_with_deleted.compacting"));
	ASSERT_FALSE(fs::exists("/tmp/half_created_sstable_with_deleted.deleted"));

	sstable_block_cache_manager::reset_instance();
}

TEST(test_sstable, empty_sstable)
{
	sstable_block_cache_manager::init_instance(5120);

	{
		sstable::pointer p = sstable::create("/tmp/empty_sstable",
						     "/tmp/empty_sstable.compacting",
						     "/tmp/empty_sstable.deleted",
						     "00000",
						     "11111",
						     0,
						     64 * 1024,
						     compressor::NONE);
		ASSERT_TRUE(p);

		ASSERT_EQ(0, p->sequence_number());
		ASSERT_EQ("", p->get_split_row());
		ASSERT_EQ("/tmp/empty_sstable", p->path());

		ASSERT_EQ(0, p->file_size());
		ASSERT_EQ(false, p->is_ref_sstable());
		ASSERT_EQ(false, p->is_ref_top_sstable());
		ASSERT_EQ(false, p->is_ref_bottom_sstable());

		{
			sstable_block_cache block;
			block_entry entry;
			std::pair<std::size_t, std::size_t> pair(0,0);
			ASSERT_FALSE(p->get_entry(block, entry, pair));
		}

		ASSERT_FALSE(fs::exists("/tmp/empty_sstable"));
		ASSERT_TRUE(fs::exists("/tmp/empty_sstable.compacting"));
		
		p->append_eof();

		ASSERT_EQ(0, p->sequence_number());
		ASSERT_EQ("", p->get_split_row());
		ASSERT_EQ("/tmp/empty_sstable", p->path());

		ASSERT_EQ((2 * sstable_block_header::SBH_LENGTH + sstable_trailer::ST_LENGTH),
			  p->file_size());
		ASSERT_EQ(false, p->is_ref_sstable());
		ASSERT_EQ(false, p->is_ref_top_sstable());
		ASSERT_EQ(false, p->is_ref_bottom_sstable());

		{
			sstable_block_cache block;
			block_entry entry;
			std::pair<std::size_t, std::size_t> pair(0,0);
			ASSERT_FALSE(p->get_entry(block, entry, pair));
		}

		ASSERT_TRUE(fs::exists("/tmp/empty_sstable"));
		ASSERT_FALSE(fs::exists("/tmp/empty_sstable.compacting"));

		ASSERT_TRUE(p->set_deleted());
		ASSERT_TRUE(fs::exists("/tmp/empty_sstable.deleted"));
	}
	ASSERT_FALSE(fs::exists("/tmp/empty_sstable"));
	ASSERT_FALSE(fs::exists("/tmp/empty_sstable.compacting"));
	ASSERT_FALSE(fs::exists("/tmp/empty_sstable.deleted"));

	sstable_block_cache_manager::reset_instance();
}

TEST(test_sstable, tiny_sstable)
{
	sstable_block_cache_manager::init_instance(5120 * 1024); // 5M

	const std::string row_prefix("tiny_sstable_");
	const std::string column_prefix("column_");
	const std::string value_prefix("1234567890abcdefg_");
	const uint64_t timestamp = 0;

	char buffer[1024 * 1024];

	const int count = 100000;

	// write
	std::size_t total_write_bytes = 0;
	{
		sstable::pointer p = sstable::create("/tmp/tiny_sstable",
						     "/tmp/tiny_sstable.compacting",
						     "/tmp/tiny_sstable.deleted",
						     "",
						     "",
						     0,
						     64 * 1024,
						     compressor::NONE);

		// append data
		for(int i=0; i<count; ++i)
		{
			std::ostringstream out;
			out.width(16);
			out.fill('0');
			out << i;

			kb::binary_oarchive<kb::network_byteorder> bout(buffer, 1024 * 1024);
			
			const char * key_data = bout.left_buffer();
			bout << row_prefix + out.str();
			bout << column_prefix + out.str();
			bout << timestamp;

			const char * value_data = bout.left_buffer();
			bout << value_prefix + out.str();

			p->append(reinterpret_cast<const key_wrapper*>(key_data),
				  reinterpret_cast<const value_wrapper*>(value_data));

			total_write_bytes += bout.used_buffer_size();
		}

		p->append_eof();

		ASSERT_EQ((row_prefix.size()
			   + column_prefix.size()
			   + sizeof(uint64_t)
			   + value_prefix.size()
			   + sizeof(uint32_t) * 3
			   + 16 * 3) * count,
			  total_write_bytes);
	}

	// read
	std::size_t total_read_bytes = 0;
	{
		sstable::pointer p = sstable::load("/tmp/tiny_sstable",
						   "/tmp/tiny_sstable",
						   "/tmp/tiny_sstable.deleted",
						   "",
						   "",
						   0,
						   ST_NORMAL_SSTABLE);


		sstable_block_cache block;
		block_entry entry;
		int i = 0;
		for(std::pair<size_t, size_t> pos(0,0);
		    p->get_entry(block, entry, pos);
		    p->advance_index(pos))
		{
			total_read_bytes += entry.m_key->bytes();
			total_read_bytes += entry.m_value->bytes();

			std::ostringstream out;
			out.width(16);
			out.fill('0');
			out << i;

			i++;

			ASSERT_TRUE((row_prefix + out.str()) == (std::string)(*entry.m_key->row()));
			ASSERT_TRUE((column_prefix + out.str()) == (std::string)(*entry.m_key->column()));
			ASSERT_EQ(0, entry.m_key->timestamp()->get());
			ASSERT_TRUE((value_prefix + out.str()) == (std::string)(*entry.m_value));
		}
		ASSERT_EQ(count, i);

	}

	ASSERT_EQ(total_write_bytes, total_read_bytes);

	// random read
	{
		sstable::pointer p = sstable::load("/tmp/tiny_sstable",
						   "/tmp/tiny_sstable",
						   "/tmp/tiny_sstable.deleted",
						   "",
						   "",
						   0,
						   ST_NORMAL_SSTABLE);


		const int count = 10000; // 随机读的次数
		// 块数和块内记录数的估计值，略大一些，令其可能越界
		const int block_number = 150;
		const int entry_number_per_block = 3000;

		srandom(time(NULL));
		for(int i = 0; i < count; ++i)
		{
			std::pair<size_t, size_t> pos;
			pos.first = (size_t)(block_number * (random() / (RAND_MAX + 1.0)));
			pos.second = (size_t)(entry_number_per_block * (random() / (RAND_MAX + 1.0)));

			sstable_block_cache block;
			block_entry entry;
			const bool ok = p->get_entry(block, entry, pos);

			if(ok)
			{
				std::cout << "(" << pos.first
					  << ", " << pos.second
					  << ") : {" << (std::string)(*entry.m_key->row())
					  << ", " << (std::string)(*entry.m_key->column())
					  << ", " << (entry.m_key->timestamp()->get())
					  << ", " << (std::string)(*entry.m_value)
					  << "}" << std::endl;
			}
			else
			{
				std::cout << "(" << pos.first
					  << ", " << pos.second
					  << ") : failed" << std::endl;
			}
			std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;
		}

		// read out of bound
		{
			std::pair<size_t, size_t> pos(1000000, 0);
			sstable_block_cache block;
			block_entry entry;
			ASSERT_FALSE(p->get_entry(block, entry, pos));
		}

		{
			std::pair<size_t, size_t> pos(0, 1000000);
			sstable_block_cache block;
			block_entry entry;
			ASSERT_TRUE(p->get_entry(block, entry, pos));
			ASSERT_EQ(1, pos.first);
			ASSERT_EQ(0, pos.second);
		}

		{
			std::pair<size_t, size_t> pos(1000000, 1000000);
			sstable_block_cache block;
			block_entry entry;
			ASSERT_FALSE(p->get_entry(block, entry, pos));
		}
	}

	sstable_block_cache_manager::reset_instance();
}

TEST(test_sstable, small_sstable)
{
	sstable_block_cache_manager::init_instance(50 * 1024 * 1024); // 50M

	const std::string row_prefix("small_sstable_");
	const std::string column_prefix("column_");
	const std::string value_prefix("1234567890abcdefg_");
	const uint64_t timestamp = 0;

	char buffer[1024 * 1024];

	const int count = 1000000;

	// write
	std::size_t total_write_bytes = 0;
	{
		TIME_EVALUATOR("creating /tmp/small_sstable ", "");
		sstable::pointer p = sstable::create("/tmp/small_sstable",
						     "/tmp/small_sstable.compacting",
						     "/tmp/small_sstable.deleted",
						     "",
						     "",
						     0,
						     64 * 1024,
						     compressor::NONE);

		// append data
		for(int i=0; i<count; ++i)
		{
			std::ostringstream out;
			out.width(16);
			out.fill('0');
			out << i;

			kb::binary_oarchive<kb::network_byteorder> bout(buffer, 1024 * 1024);
			
			const char * key_data = bout.left_buffer();
			bout << row_prefix + out.str();
			bout << column_prefix + out.str();
			bout << timestamp;

			const char * value_data = bout.left_buffer();
			bout << value_prefix + out.str();

			p->append(reinterpret_cast<const key_wrapper*>(key_data),
				  reinterpret_cast<const value_wrapper*>(value_data));

			total_write_bytes += bout.used_buffer_size();
		}

		p->append_eof();

		ASSERT_EQ((row_prefix.size()
			   + column_prefix.size()
			   + sizeof(uint64_t)
			   + value_prefix.size()
			   + sizeof(uint32_t) * 3
			   + 16 * 3) * count,
			  total_write_bytes);

		std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;
	}
	std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;

	// read
	std::size_t total_read_bytes = 0;
	{
		TIME_EVALUATOR("seq-read /tmp/small_sstable ", "");
		sstable::pointer p = sstable::load("/tmp/small_sstable",
						   "/tmp/small_sstable",
						   "/tmp/small_sstable.deleted",
						   "",
						   "",
						   0,
						   ST_NORMAL_SSTABLE);


		sstable_block_cache block;
		block_entry entry;
		int i = 0;
		for(std::pair<size_t, size_t> pos(0,0);
		    p->get_entry(block, entry, pos);
		    p->advance_index(pos))
		{
			total_read_bytes += entry.m_key->bytes();
			total_read_bytes += entry.m_value->bytes();

			std::ostringstream out;
			out.width(16);
			out.fill('0');
			out << i;

			i++;

			ASSERT_TRUE((row_prefix + out.str()) == (std::string)(*entry.m_key->row()));
			ASSERT_TRUE((column_prefix + out.str()) == (std::string)(*entry.m_key->column()));
			ASSERT_EQ(0, entry.m_key->timestamp()->get());
			ASSERT_TRUE((value_prefix + out.str()) == (std::string)(*entry.m_value));
		}
		ASSERT_EQ(count, i);

		std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;
	}
	std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;

	ASSERT_EQ(total_write_bytes, total_read_bytes);

	// random read
	{
		TIME_EVALUATOR("random-read /tmp/small_sstable ", "");
		sstable::pointer p = sstable::load("/tmp/small_sstable",
						   "/tmp/small_sstable",
						   "/tmp/small_sstable.deleted",
						   "",
						   "",
						   0,
						   ST_NORMAL_SSTABLE);


		const int count = 10000; // 随机读的次数
		// 块数和块内记录数的估计值，略大一些，令其可能越界
		const int block_number = 2000;
		const int entry_number_per_block = 3000;

		srandom(time(NULL));
		for(int i = 0; i < count; ++i)
		{
			std::pair<size_t, size_t> pos;
			pos.first = (size_t)(block_number * (random() / (RAND_MAX + 1.0)));
			pos.second = (size_t)(entry_number_per_block * (random() / (RAND_MAX + 1.0)));

			sstable_block_cache block;
			block_entry entry;
			const bool ok = p->get_entry(block, entry, pos);

			if(ok)
			{
				std::cout << "(" << pos.first
					  << ", " << pos.second
					  << ") : {" << (std::string)(*entry.m_key->row())
					  << ", " << (std::string)(*entry.m_key->column())
					  << ", " << (entry.m_key->timestamp()->get())
					  << ", " << (std::string)(*entry.m_value)
					  << "}" << std::endl;
			}
			else
			{
				std::cout << "(" << pos.first
					  << ", " << pos.second
					  << ") : failed" << std::endl;
			}
		}
		std::cout << sstable_block_cache_manager::instance().to_string() << std::endl;

		// read out of bound
		{
			std::pair<size_t, size_t> pos(1000000, 0);
			sstable_block_cache block;
			block_entry entry;
			ASSERT_FALSE(p->get_entry(block, entry, pos));
		}

		{
			std::pair<size_t, size_t> pos(0, 1000000);
			sstable_block_cache block;
			block_entry entry;
			ASSERT_TRUE(p->get_entry(block, entry, pos));
			ASSERT_EQ(1, pos.first);
			ASSERT_EQ(0, pos.second);
		}

		{
			std::pair<size_t, size_t> pos(1000000, 1000000);
			sstable_block_cache block;
			block_entry entry;
			ASSERT_FALSE(p->get_entry(block, entry, pos));
		}
	}

	sstable_block_cache_manager::reset_instance();
}
